/*******************************************************************************
 * Package: com.song.bigdata.batch
 * Type:    FlinkBatchTxt
 * Date:    2022-10-22 21:13
 *
 * Copyright (c) 2022 HUANENG GUICHENG TRUST CORP.,LTD All Rights Reserved.
 *
 * You may not use this file except in compliance with the License.
 *******************************************************************************/
package com.song.bigdata.batch;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.AggregateOperator;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.FlatMapOperator;
import org.apache.flink.api.java.operators.UnsortedGrouping;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

/**
 * 功能描述：
 *  没启动起来  一直报The TaskExecutor is shutting down. 烦死了  真倒霉
 * @author Songxianyang
 * @date 2022-10-22 21:13
 */
public class FlinkBatchTxt {
    public static void main(String[] args) throws Exception {
        // 获取环境
        ExecutionEnvironment environment = ExecutionEnvironment.getExecutionEnvironment();
        // 获取文件
        DataSource<String> textFile = environment.readTextFile("steve-big-data/file/batchFlie.txt");
        // 解析文件  数据格式转换

        FlatMapOperator<String, Tuple2<String, Long>> returns = textFile.flatMap((String s, Collector<Tuple2<String, Long>> out) -> {
            String[] names = s.split(" ");
            for (String name : names) {
                out.collect(Tuple2.of(name, 1L));
            }
        }).returns(Types.TUPLE(Types.STRING, Types.LONG));

        UnsortedGrouping<Tuple2<String, Long>> groupBy =
                returns.groupBy(0);
        AggregateOperator<Tuple2<String, Long>> sum =
                groupBy.sum(1);
        sum.print();
    }
}
